import threading
import multiprocessing
from functools import partial


class DragonDynamicThreadPool:
    def __init__(self, max_workers: int = None):
        """
        初始化动态线程池
        :param max_workers: 最大线程数，None表示使用默认值（通常为CPU核心数*5）
        """
        # 初始化线程池，max_workers控制最大并发数
        self.pool = ThreadPoolExecutor(max_workers=max_workers)
        self.futures = []  # 存储所有提交的任务future

    def add_thread(self, func: callable, fun_args: list):
        """添加线程任务"""
        future = self.pool.submit(func, *fun_args)
        self.futures.append(future)
        return future

    def add_task(self, func: callable, fun_args: list):
        """添加任务（与add_thread功能一致）"""
        return self.add_thread(func, fun_args)

    def start(self):
        """等待所有任务完成"""
        # 等待所有任务执行完毕
        for future in self.futures:
            try:
                # 获取结果（会阻塞直到任务完成）
                future.result()
            except Exception as e:
                # 捕获任务执行中的异常
                print(f"任务执行出错: {e}")
        # 关闭线程池
        self.pool.shutdown(wait=True)

    def run(self):
        """启动并等待所有任务完成"""
        self.start()



class DragonDynamicProcessPool:
    """
    多进程之间不能共享同一个对象
    """
    def __init__(self):
        """
        多进程之间不能共享同一个对象
        """
        # self.pool = multiprocessing.Pool(processes=num_threads)
        self.pool = None
        self.threads = []

    def apply_async(self, func: callable, fun_args: list):
        self.threads.append((func, fun_args))
        # self.pool.apply_async(func, fun_args)

    def add_thread(self, func: callable, fun_args: list):
        self.apply_async(func, fun_args)

    def add_task(self, func: callable, fun_args: list):
        self.apply_async(func, fun_args)

    def start(self):
        self.pool = multiprocessing.Pool(processes=len(self.threads))
        for func, fun_args in self.threads:
            self.pool.apply_async(func, fun_args)
        self.pool.close()
        self.pool.join()
    def run(self):
        self.start()

from concurrent.futures import ThreadPoolExecutor
from functools import partial
import threading

class DragonFixedThreadPool:
    def __init__(self, num_threads: int):
        """
        初始化线程池
        :param num_threads: 线程数量
        """
        self.pool = ThreadPoolExecutor(max_workers=num_threads)
        self.futures = []  # 用于存储异步任务的future对象

    def map(self, func: callable, iterable_arg_list: list, other_fun_args: dict):
        """
        批量处理任务，类似多进程版本的map方法
        :param func: 要执行的函数
        :param iterable_arg_list: 可迭代的参数列表
        :param other_fun_args: 函数的其他关键字参数
        :return: 函数执行结果列表
        """
        partial_my_function = partial(func, **other_fun_args)
        return list(self.pool.map(partial_my_function, iterable_arg_list))

    def apply_async(self, func: callable, fun_args: list):
        """
        异步提交单个任务
        :param func: 要执行的函数
        :param fun_args: 函数的位置参数列表
        """
        future = self.pool.submit(func, *fun_args)
        self.futures.append(future)
        return future

    def add_thread(self, func: callable, fun_args: list):
        """添加线程任务（与apply_async功能一致）"""
        return self.apply_async(func, fun_args)

    def add_task(self, func: callable, fun_args: list):
        """添加任务（与apply_async功能一致）"""
        return self.apply_async(func, fun_args)

    def start(self):
        """等待所有异步任务完成"""
        # 等待所有future完成
        for future in self.futures:
            future.result()  # 获取结果会阻塞直到任务完成
        self.pool.shutdown(wait=True)

    def run(self):
        """启动并等待所有任务完成"""
        self.start()


class DragonFixedProcessPool:
    def __init__(self, num_threads: int):
        self.pool = multiprocessing.Pool(processes=num_threads)

    def map(self, func: callable, iterable_arg_list: list, other_fun_args: dict):
        """
        要求函数只用第一个参数是遍历list。
        :param func:
        :param fun_args:
        :return:
        """
        partial_my_function = partial(func, **other_fun_args)
        return self.pool.map(partial_my_function, iterable_arg_list)

    def apply_async(self, func: callable, fun_args: list):
        self.pool.apply_async(func, fun_args)

    def add_thread(self, func: callable, fun_args: list):
        self.apply_async(func, fun_args)

    def add_task(self, func: callable, fun_args: list):
        self.apply_async(func, fun_args)

    def start(self):
        self.pool.close()
        self.pool.join()
    def run(self):
        self.start()